package cn.itcast.flink.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @author lilulu
 */
public class StreamSourceOrderDemo {
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Order {
        private String id;
        private Integer userId;
        private Double money;
        private long orderTime;
    }

    /**
     * 自定义数据源，继承抽象类：RichParallelSourceFunction，并行的和富有的
     */
    public static class OrderSource extends RichParallelSourceFunction<Order> {
        private boolean isRunning = true;

        //产生数据，从数据源源源不断的产生数据
        public void run(SourceContext<Order> sourceContext) throws Exception {
            Random random = new Random();
            while (true) {
                Order order = new Order(UUID.randomUUID().toString(),
                        random.nextInt(2),
                        (double) random.nextInt(100),
                        System.currentTimeMillis());
                //发送订单数据
                sourceContext.collect(order);
                //每隔5秒产生数据
                TimeUnit.SECONDS.sleep(5);
            }
        }

        // 取消从数据源加载数据
        public void cancel() {
            isRunning = false;
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Order> orderDataStreamSource = env.addSource(new OrderSource());
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        orderDataStreamSource.printToErr();
        // 5. 触发执行-execute
        env.execute("StreamSourceOrderDemo");
    }
}